/*******************************************************************************
 * Copyright (c) 2012, 2017 IBM Corp.
 *
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * and Eclipse Distribution License v1.0 which accompany this distribution.
 *
 * The Eclipse Public License is available at
 *   http://www.eclipse.org/legal/epl-v10.html
 * and the Eclipse Distribution License is available at
 *   http://www.eclipse.org/org/documents/edl-v10.php.
 *
 * Contributors:
 *    Ian Craggs - initial contribution
 *******************************************************************************/
#include <sqlite3.h>
#include <time.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "MQTTClient.h"
#include "cJSON.h"
#include "shmem.h"
#include "msg_queue_peer.h"
#define ADDRESS "tcp://192.168.51.133:1883"
#define CLIENTIDS "ClientSub"

#define TOPICS "Down"
#define TOPICP "Up"

#define PAYLOAD "Hello World!"
#define QOS 1
#define TIMEOUT 10000L
static struct shm_param para;
sqlite3 *db;
int report_type, report_period;
char mqtt_server_addr[32];
int mqtt_server_port;
char mb_dev_addr[32];
int mb_dev_port;
char up_buf[1024]; // 要发送的msg
union val_t
{
    int b_val;   // bool类型存储空间
    int i_val;   // 整形值存储空间
    float f_val; // 浮点值存储空间
};

struct std_node
{
    int key;             // 唯一键值
    int type;            // 数据点类型
    int dev_type;        // 数据点属于哪个设备，根据网关支持的设备自行定义
    union val_t old_val; // 变化上报后需要更新旧值
    union val_t new_val; // 从共享内存取出最新数据，放到new_val中
    int ret;             // 默认为-1，采集成功后设置为0，采集失败再置-1
};
volatile MQTTClient_deliveryToken deliveredtoken;

void delivered(void *context, MQTTClient_deliveryToken dt)
{
    printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}

int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    int i;
    char *payloadptr;
    // int msgtype;
    // char order[32];
    
    printf("Message arrived\n");
    printf("     topic: %s\n", topicName);
    printf("   message: \n");

    payloadptr = message->payload;
    // json的反序列化
    cJSON *root = cJSON_Parse(payloadptr);
    // 一定要判断是否是标准的JSON格式（即object类型，排除掉那些因为是JSON其他的基本数据类型而被认为成是JSON的情况）
    if ((NULL == root) || (root->type != cJSON_Object))
    {
        printf("parse err\n");
        cJSON_Delete(root);
        return 1;
    }

    // 逐个打印解析出来的节点信息
    cJSON *name = cJSON_GetObjectItem(root, "name");
    printf("name = %s\n", name->valuestring);
    cJSON *age = cJSON_GetObjectItem(root, "age");
    printf("age = %d\n", age->valueint);
    cJSON *msg = cJSON_GetObjectItem(root, "msg");
    printf("msg = %s\n", msg->valuestring);

    // 指令处理
    // 把指令写到消息队列里

    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    // 释放root
    cJSON_Delete(root);
    return 1;
}

void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}

// 数据库存储线程
void *sqlite_pthread(void *arg)
{
    struct std_node *saddr = (struct std_node *)arg;
    char arr[32];
    char sql[128];
    char *errmsg;
    time_t current_time;
    
    while (1)
    {
        sleep(60);
        
        for (int i = 0; i < 12; i++)
        {
            sprintf(arr, "%d", saddr[i].new_val.i_val);
            sprintf(sql, "insert into data values(%d,%d,'%s')", saddr[i].key, time(&current_time), arr);

            if (sqlite3_exec(db, sql, NULL, NULL, &errmsg) != SQLITE_OK)
            {
                fprintf(stderr, "insert err:%s\n", errmsg);
                return NULL;
            }
            else
            {
                printf("数据库内容更新成功\n");
            }
        }
    }
}

int main(int argc, char *argv[])
{
    /*************************************************************共享内存**********************************************************/

    int ret = shm_init(&para, "shm_test", 1024); // 注意进程双方共享内存的大小一致
    if (ret < 0)
    {
        printf("共享内存初始化失败\n");
        return -1;
    }
    int *addr = (int *)shm_getaddr(&para);
    if (addr == NULL)
    {
        printf("共享内存获取地址失败\n");
        return -1;
    }
    *addr = 12;
    struct std_node *saddr = (struct std_node *)(addr + sizeof(int));

    /************************************************************解析点表**********************************************************/

    FILE *file = fopen("node.json", "r"); // 读取文件内容
    if (NULL == file)
    {
        perror("read err\n");

        return -1;
    }

    char buf[3000] = {0};

    size_t len = fread(buf, 1, 3000, file);
    if (len <= 0)
    {
        perror("read err");
        return -1;
    }
    // printf("buf = %s\n", buf);
    // json的反序列化
    cJSON *root = cJSON_Parse(buf);
    // 一定要判断
    if (NULL == root)
    {
        printf("parse err\n");
        cJSON_Delete(root);
        fclose(file);
        return -1;
    }

    // 获取root上节点
    cJSON *report = cJSON_GetObjectItem(root, "report");
    cJSON *repo_type = cJSON_GetObjectItem(report, "type");
    cJSON *repo_period = cJSON_GetObjectItem(report, "period");
    report_type = repo_type->valuestring;
    report_period = repo_period->valueint;
    cJSON *mqtt_ser = cJSON_GetObjectItem(root, "mqtt_server");
    cJSON *mqtt_addr = cJSON_GetObjectItem(mqtt_ser, "addr");
    cJSON *mqtt_port = cJSON_GetObjectItem(mqtt_ser, "port");
    strcpy(mqtt_server_addr, mqtt_addr->valuestring);
    mqtt_server_port = mqtt_port->valueint;

    cJSON *mb_dev = cJSON_GetObjectItem(root, "mb_dev");
    cJSON *mb_addr = cJSON_GetObjectItem(mb_dev, "addr");
    cJSON *mb_port = cJSON_GetObjectItem(mb_dev, "port");
    strcpy(mb_dev_addr, mb_addr->valuestring);
    mb_dev_port = mb_port->valueint;

    cJSON *stm = cJSON_GetObjectItem(root, "stm32");
    cJSON *data = cJSON_GetObjectItem(stm, "data");
    int count = cJSON_GetArraySize(data);
    cJSON *tmp;

    for (int i = 0; i < count; i++)
    {
        tmp = cJSON_GetArrayItem(data, i);
        cJSON *key = cJSON_GetObjectItem(tmp, "key");
        cJSON *type = cJSON_GetObjectItem(tmp, "type");
        saddr[i].key = key->valueint;
        saddr[i].type = type->valueint;
        saddr[i].dev_type = 0;
        saddr[i].ret = -1;
    }
 
    cJSON *mdbus = cJSON_GetObjectItem(root, "modbus");
    data = cJSON_GetObjectItem(mdbus, "data");
    count = cJSON_GetArraySize(data);
   
    for (int i = 0; i < count; i++)
    {

        tmp = cJSON_GetArrayItem(data, i);
        cJSON *key = cJSON_GetObjectItem(tmp, "key");
        cJSON *type = cJSON_GetObjectItem(tmp, "type");
        saddr[i+6].key = key->valueint;
        saddr[i+6].type = type->valueint;
        saddr[i+6].dev_type = 1;
    }
   
    cJSON_Delete(root);

    /*******************************************************************创建数据库**************************************************/

    // 打开或创建数据库
    if (sqlite3_open("history.db", &db) != SQLITE_OK)
    {
        fprintf(stderr, "open err:%s\n", sqlite3_errmsg(db));
        return -1;
    }
    printf("open okk\n");

    char *errmsg = NULL;
    // 表： key(设备唯一标识)  time(时间戳)    value(数据值)
    if (sqlite3_exec(db,
                     "CREATE TABLE IF NOT EXISTS data ("
                     "key INTEGER, "
                     "time INTEGER, "
                     "value TEXT)",
                     NULL, NULL, &errmsg) != SQLITE_OK)
    {
        fprintf(stderr, "create err:%s\n", errmsg);
        return -1;
    }
    printf("create data table ok\n");

    if (sqlite3_exec(db,
                     "CREATE TABLE IF NOT EXISTS user ("
                     "name TEXT, "
                     "password TEXT)",
                     NULL, NULL, &errmsg) != SQLITE_OK)
    {
        fprintf(stderr, "create err:%s\n", errmsg);
        return -1;
    }
    printf("create user table ok\n");

    /*******************************************************************数据存储线程**************************************************/

    pthread_t tid0;
    // 创建线程
    if (pthread_create(&tid0, NULL, sqlite_pthread, saddr))
    {
        printf("create sqlite thread err\n");
        return -1;
    }
    else
    {
        printf("数据库存储线程创建成功\n");
    }
    // 主线程分离
    pthread_detach(tid0);

 
    /*******************************************************************连接MQTT服务器**************************************************/

    MQTTClient clients;
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;

    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    MQTTClient_deliveryToken token;

    int rc;
 
    // 创建客户端，并且指定客户端连接的mqtt服务器地址和客户端ID
    MQTTClient_create(&clients, ADDRESS, CLIENTIDS,
                      MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;
    // 设置回调接口，只需要关注msgarrvd：消息到达后，会自动调用这个接口（相当于开了个线程负责接收消息）
    MQTTClient_setCallbacks(clients, NULL, connlost, msgarrvd, delivered);
   
    if ((rc = MQTTClient_connect(clients, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {        
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    
    printf("Subscribing to topic %s\nfor client %s using QoS%d\n\n"
           "Press Q<Enter> to quit\n\n",
           TOPICS, CLIENTIDS, QOS);
    MQTTClient_subscribe(clients, TOPICS, QOS);
 struct std_node *node_arr = saddr;    // 指向共享内存中节点缓存数组头
                                          // 创建序列化树
    cJSON *root1 = cJSON_CreateObject();  // 创建JSON对象
    cJSON *stm32 = cJSON_CreateObject();  // 创建stm32对象
    cJSON *modbus = cJSON_CreateObject(); // 创建modbus对象
    cJSON_AddItemToObject(root1, "stm32", stm32);
    cJSON_AddItemToObject(root1, "modbus", modbus);
    cJSON *data1 = cJSON_CreateArray();
    cJSON *data2 = cJSON_CreateArray();
    cJSON_AddItemToObject(stm32, "data", data1);
    cJSON_AddItemToObject(modbus, "data", data2);
    /******************************************数据上报*********************************************/
    //  刷新上报---不上报
    if (report_type == 0)
    {
        while(1){

        }
        
    }
    /******************************变化上报***********************************/
    else if (report_type == 1)
    {
        int i=0;
        // 先将所有的数据上报一次
        for (i = 0; i < *addr; i++)
        {
            cJSON *item = cJSON_CreateObject();
            cJSON_AddItemToObject(item, "key", cJSON_CreateNumber(node_arr[i].key));
            //"val"
            if (node_arr[i].type == 1 || node_arr[i].type == 2)
            {
                cJSON *temp = cJSON_CreateNumber(node_arr[i].new_val.i_val);
                cJSON_AddItemToObject(item, "val", temp);
                //"设备属于哪个部分"
                if (node_arr[i].dev_type == 0)
                {

                    cJSON_AddItemToArray(data1, item);
                }
                else if (node_arr[i].dev_type == 1)
                {
                    cJSON_AddItemToArray(data2, item);
                }
            }
            else if (node_arr[i].type == 3)
            {
                cJSON *temp = cJSON_CreateNumber(node_arr[i].new_val.f_val);
                cJSON_AddItemToObject(item, "val", temp);
                //"设备属于哪个部分"
                if (node_arr[i].dev_type == 0)
                {
                    cJSON_AddItemToArray(data1, item);
                }
                else if (node_arr[i].dev_type == 1)
                {
                    cJSON_AddItemToArray(data2, item);
                }
            }
        }
        char *up_buf = cJSON_Print(root1);
        printf("%s", up_buf);
        pubmsg.payload = up_buf;
        pubmsg.payloadlen = (int)strlen(up_buf);
        pubmsg.qos = QOS;
        pubmsg.retained = 0;
        MQTTClient_publishMessage(clients, TOPICP, &pubmsg, &token);
        rc = MQTTClient_waitForCompletion(clients, token, TIMEOUT);
        memset(up_buf, 0, sizeof(up_buf));
        free(up_buf);
        /*********************************第一次上报全部数据点结束*******************************************/
        while (1)
        {
            int i = 0;
            for (i = 0; i < *addr; i++)
            {

                if (node_arr[i].old_val.i_val != node_arr[i].new_val.i_val)
                {
                    cJSON *item = cJSON_CreateObject();
                    cJSON_AddItemToObject(item, "key", cJSON_CreateNumber(node_arr[i].key));
                    //"val"
                    if (node_arr[i].type == 1 || node_arr[i].type == 2)
                    {
                        cJSON *temp = cJSON_CreateNumber(node_arr[i].new_val.i_val);
                        cJSON_AddItemToObject(item, "val", temp);
                        //"设备属于哪个部分"
                        if (node_arr[i].dev_type == 0)
                        {

                            cJSON_AddItemToArray(data1, item);
                        }
                        else if (node_arr[i].dev_type == 1)
                        {
                            cJSON_AddItemToArray(data2, item);
                        }
                    }
                    else if (node_arr[i].type == 3)
                    {
                        cJSON *temp = cJSON_CreateNumber(node_arr[i].new_val.f_val);
                        cJSON_AddItemToObject(item, "val", temp);
                        //"设备属于哪个部分"
                        if (node_arr[i].dev_type == 0)
                        {
                            cJSON_AddItemToArray(data1, item);
                        }
                        else if (node_arr[i].dev_type == 1)
                        {
                            cJSON_AddItemToArray(data2, item);
                        }
                    }
                }
                char *up_buf = cJSON_Print(root1);
                printf("%s", up_buf);
                pubmsg.payload = up_buf;
                pubmsg.payloadlen = (int)strlen(up_buf);
                pubmsg.qos = QOS;
                pubmsg.retained = 0;
                MQTTClient_publishMessage(clients, TOPICP, &pubmsg, &token);
                rc = MQTTClient_waitForCompletion(clients, token, TIMEOUT);
                memset(up_buf, 0, sizeof(up_buf));
                free(up_buf);
            }
        }
    }

    // 周期上报
    else if (report_type == 2)
    {
        while (1)
        {
            int i = 0;
            // 序列化
            for (i = 0; i < *addr; i++)
            {
                cJSON *item = cJSON_CreateObject();
                cJSON_AddItemToObject(item, "key", cJSON_CreateNumber(node_arr[i].key));
                //"val"
                if (node_arr[i].type == 1 || node_arr[i].type == 2)
                {
                    cJSON *temp = cJSON_CreateNumber(node_arr[i].new_val.i_val);
                    cJSON_AddItemToObject(item, "val", temp);
                    //"设备属于哪个部分"
                    if (node_arr[i].dev_type == 0)
                    {

                        cJSON_AddItemToArray(data1, item);
                    }
                    else if (node_arr[i].dev_type == 1)
                    {
                        cJSON_AddItemToArray(data2, item);
                    }
                }
                else if (node_arr[i].type == 3)
                {
                    cJSON *temp = cJSON_CreateNumber(node_arr[i].new_val.f_val);
                    cJSON_AddItemToObject(item, "val", temp);
                    //"设备属于哪个部分"
                    if (node_arr[i].dev_type == 0)
                    {
                        cJSON_AddItemToArray(data1, item);
                    }
                    else if (node_arr[i].dev_type == 1)
                    {
                        cJSON_AddItemToArray(data2, item);
                    }
                }
            }
            char *up_buf = cJSON_Print(root1);
            printf("%s", up_buf);
            pubmsg.payload = up_buf;
            pubmsg.payloadlen = (int)strlen(up_buf);
            pubmsg.qos = QOS;
            pubmsg.retained = 0;
            MQTTClient_publishMessage(clients, TOPICP, &pubmsg, &token);
            rc = MQTTClient_waitForCompletion(clients, token, TIMEOUT);
            memset(up_buf, 0, sizeof(up_buf));
            free(up_buf);
            sleep(report_period);
        }
    }

    cJSON_Delete(root1);
    MQTTClient_unsubscribe(clients, TOPICS);
    MQTTClient_unsubscribe(clients, TOPICP);
    MQTTClient_disconnect(clients, 10000);
    MQTTClient_destroy(&clients);
    return rc;
}
